package com.spc.redis;


import com.spc.redis.configurer.RedisClientProperties;
import com.spc.redis.serializer.*;
import com.spc.redis.support.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.lang.NonNull;
import org.springframework.util.Assert;

import java.io.IOException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * client for redis with {@link }
 *
 * @author alex
 * @version v1.0 2021/3/17
 */
public class RedisHighClient {

    private final RedisTemplate<String, String> redisTemplate;

    private final Serializer serializer;

    private final RedisClientProperties redisClientProperties;

    public static final Logger logger = LoggerFactory.getLogger(RedisHighClient.class);


    public RedisHighClient(RedisTemplate redisTemplate, RedisClientProperties redisClientProperties) {
        this.redisTemplate = redisTemplate;
        this.redisClientProperties = redisClientProperties;
        this.serializer = findCandidateSerializer(redisClientProperties.getSerializerType());
        this.redisTemplate.setKeySerializer(new StringRedisSerializer());
    }

    /**
     * SET
     *
     * @param key
     * @param value
     * @return
     */
    public boolean set(String key, String value) {
        redisTemplate.opsForValue().set(key, value);
        return true;
    }

    /**
     * SET 对象
     *
     * @param key
     * @param value
     * @param <T>
     * @return
     */
    public <T> boolean set(String key, T value) {
        try {
            redisTemplate.opsForValue().set(key, this.serializer.serialize(value));
            return true;
        } catch (SerializeException e) {
            logger.error(e.getLocalizedMessage());
        }
        return false;
    }

    /**
     * SETEX
     *
     * @param key
     * @param value
     * @param ttlDuration
     * @return
     */
    public boolean setex(String key, String value, Duration ttlDuration) {
        redisTemplate.opsForValue().set(key, value, ttlDuration.getSeconds(),
                TimeUnit.SECONDS);
        return true;
    }

    /**
     * SETEX
     *
     * @param key
     * @param value
     * @param ttlDuration
     * @param <T>
     * @return
     */
    public <T> boolean setex(String key, T value, Duration ttlDuration) {
        try {
            redisTemplate.opsForValue().set(key, this.serializer.serialize(value), ttlDuration.getSeconds(),
                    TimeUnit.SECONDS);
            return true;
        } catch (SerializeException e) {
            logger.error(e.getLocalizedMessage());
        }
        return false;
    }

    /**
     * 使用默认过期时间
     *
     * @param key
     * @param value
     * @param <T>
     * @return
     */
    public <T> boolean setex(String key, T value) {
        return setex(key, value, redisClientProperties.getTtl());
    }

    /**
     * HMSET
     * 设置整个 hash
     *
     * @param key
     * @param value
     * @return
     */
    public boolean setHash(String key, Map<String, Object> value) {
        redisTemplate.opsForHash().putAll(key, value);
        return true;
    }

    /**
     * HMSET
     * 设置 hash 的键值
     *
     * @param key
     * @param hashKey
     * @param value
     * @param <T>
     * @return
     */
    public <T> boolean setHash(String key, String hashKey, T value) {
        try {
            redisTemplate.opsForHash().put(key, hashKey, this.serializer.serialize(value));
            return true;
        } catch (SerializeException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * HGETALL
     * 获取整个 hash 对象
     *
     * @param key
     * @return
     */
    public Map<Object, Object> getHash(String key) {
        return redisTemplate.opsForHash().entries(key);
    }

    /**
     * HGET
     * <p>
     * 获取某个键的值对象
     *
     * @param key
     * @param hashKey
     * @param valueType
     * @param <T>
     * @return
     */
    public <T> T getHash(String key, String hashKey, Class<T> valueType) {
        String value = String.valueOf(redisTemplate.opsForHash().get(key, hashKey));
        try {
            return this.serializer.deserialize(value, valueType);
        } catch (SerializeException e) {
            logger.error(e.getLocalizedMessage());
        }
        return null;
    }

    /**
     * LPUSH
     *
     * @param key
     * @param value
     * @param <T>
     * @return
     */
    public <T> boolean push2List(String key, T value) {
        try {
            redisTemplate.opsForList().leftPush(key, this.serializer.serialize(value));
            return true;
        } catch (SerializeException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * RPUSH
     *
     * @param key
     * @param value
     * @param <T>
     * @return
     */
    public <T> boolean rpush2List(String key, T value) {
        try {
            redisTemplate.opsForList().rightPush(key, this.serializer.serialize(value));
            return true;
        } catch (SerializeException e) {
            e.printStackTrace();
        }
        return false;
    }


    /**
     * add to SET
     * <p>
     * SADD
     *
     * @param key
     * @param value
     * @param <T>
     * @return
     */
    public <T> boolean add2Set(String key, T value) {
        try {
            redisTemplate.opsForSet().add(key, this.serializer.serialize(value));
            return true;
        } catch (SerializeException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * setex key
     *
     * @param key
     * @param ttl
     * @return
     */
    public boolean expireKey(String key, Duration ttl) {
        return redisTemplate.expire(key, ttl.getSeconds(), TimeUnit.SECONDS);
    }

    /**
     * delete key
     *
     * @param key
     * @return
     */
    public boolean deleteKey(String key) {
        Assert.notNull(key, "key must not be null");
        return redisTemplate.delete(key);
    }


    /**
     * 简单的字符串类型， 对拿到的字符串进行反序列化转换等
     * <p>
     * get for object
     *
     * @param key
     * @param valueType
     * @param <T>
     * @return
     */
    public <T> T getForObject(String key, Class<T> valueType) {
        String value = redisTemplate.opsForValue().get(key);
        if (value != null) {
            try {
                return this.serializer.deserialize(value, valueType);
            } catch (SerializeException e) {
                logger.error(e.getLocalizedMessage());
            }
        }
        return null;
    }


    /**
     * TESTME
     * list 存储的 redis数据
     *
     * @param key
     * @param valueType
     * @return
     */
    public <T> List<T> getForList(String key, Class<T> valueType) {
        List<String> list = redisTemplate.opsForList().range(key, 0, -1);
        if (list != null && list.size() > 0) {
            return list.stream().map(value -> {
                try {
                    return this.serializer.deserialize(value, valueType);
                } catch (SerializeException e) {
                    e.printStackTrace();
                }
                return null;
            }).filter(Objects::nonNull).collect(Collectors.toList());
        }
        return new ArrayList<>(1);
    }

    /**
     * HSCAN
     *
     * @param key     hash key
     * @param pattern key pattern
     * @param limit
     * @return
     */
    public Collection<Pair<String, String>> hscan(String key, String pattern, int limit) {
        List<Pair<String, String>> ret = new ArrayList<>(1);
        try {
            Cursor<Map.Entry<Object, Object>> cursor = redisTemplate.opsForHash()
                    .scan(key, ScanOptions.scanOptions().match(pattern).count(limit).build());
            while (cursor.hasNext()) {
                Map.Entry<Object, Object> entry = cursor.next();
                String hashKey = String.valueOf(entry.getKey());
                ret.add(Pair.of(hashKey, String.valueOf(entry.getValue())));
            }
            //关闭scan
            cursor.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return ret;
    }

    /**
     * SSCAN
     *
     * @param key     set 的名称
     * @param pattern set 中值匹配的模式
     * @param limit
     * @return
     */
    public Collection<String> sscan(String key, String pattern, int limit) {
        Cursor<String> cursor = null;
        List<String> ret = new ArrayList<>(1);
        try {
            cursor = redisTemplate.opsForSet().scan(key,
                    ScanOptions.scanOptions().match(pattern).count(limit).build());
            while (cursor.hasNext()) {
                ret.add(cursor.next());
            }
            cursor.close();
        } catch (IOException e) {
            logger.error(e.getLocalizedMessage());
        } finally {
            if (cursor != null) {
                try {
                    cursor.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return ret;
    }

    /**
     * 发送订阅消息
     *
     * @param channel
     * @param msg
     */
    public <T> void publish(String channel, T msg) {
        try {
            String msgContent = this.serializer.serialize(msg);
            redisTemplate.convertAndSend(channel, this.serializer.serialize(msgContent));
            logger.debug("[redis publish] channel:{},msg:{} ", channel, msgContent);
        } catch (SerializeException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅某个channel
     *
     * @param messageListener
     * @param channel
     */
    public void subscribe(@NonNull MessageListener messageListener, @NonNull String channel) {
        redisTemplate.execute(connection -> {
            connection.subscribe(messageListener, channel.getBytes());
            return null;

        }, true);
    }

    public RedisTemplate redisTemplate() {
        return redisTemplate;
    }


    private Serializer findCandidateSerializer(String type) {
        switch (type) {
            case GsonSerializer.TYPE:
                return new GsonSerializer();
            case FastJsonSerializer.TYPE:
                return new FastJsonSerializer();
            case JacksonSerializer.TYPE:
                return new JacksonSerializer();
            case HutoolJsonSerializer.TYPE:
            default:
                return new HutoolJsonSerializer();
        }
    }

}
